/*
 * 文件名：InitiativeSendHandler.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：yuanze
 * 修改时间：2014年11月30日
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.disruptor;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.realstream.server.model.BaseNetPackage;
import com.ailk.bigdata.etl.realstream.server.model.RealDataDispatcher;
import com.ailk.bigdata.etl.realstream.server.model.SocketInfo;
import com.ailk.bigdata.etl.realstream.server.model.ValueEvent;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;
import com.lmax.disruptor.WorkHandler;

/**
 * @description TODO
 * @author [yuanze]
 * @version [版本号,2014年11月30日]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class InitiativeSendHandler implements WorkHandler<ValueEvent>
{
	private static Logger logger = LoggerFactory.getLogger(InitiativeSendHandler.class);

	/**
	 * @see com.lmax.disruptor.WorkHandler#onEvent(java.lang.Object)
	 */
	@Override
	public void onEvent(ValueEvent event) throws Exception
	{
		BaseNetPackage dataPackage = event.getDataPackage();
		long onlyCode = Long.valueOf(dataPackage.getDataType());
		RealDataDispatcher dispatcher = RealStreamTool.DISPATCH_CAHCE.get(onlyCode);
		// 获得主动推送远程地址 首次进入建立channel连接
		List<SocketInfo> socketInfoList = dispatcher.getRealStreamUnit().getSocketInfo();
		for (SocketInfo socketInfo : socketInfoList)
		{
			String key = socketInfo.getSocketIp() + socketInfo.getSocketPort();
			// if (!dispatcher.getSEND_CHANNEL_CACHE().contains(key))
			// {
			// Channel channel = new
			// StreamDataInitiativeSendClient(socketInfo.getSocketIp(),
			// socketInfo.getSocketPort()).run();
			// dispatcher.getSEND_CHANNEL_CACHE().put(key, channel);
			// }
		}

		boolean sendMsgIsNormal = true;
		// for (Map.Entry<String, Channel> channelKey :
		// dispatcher.getSEND_CHANNEL_CACHE().entrySet())
		// {
		// Channel channel = channelKey.getValue();
		// if (channel == null || !channel.isActive() || !channel.isWritable())
		// {
		// dispatcher.getSEND_CHANNEL_CACHE().remove(channelKey.getKey());
		// sendMsgIsNormal = false;
		// }
		// else
		// {
		// channel.writeAndFlush(dataPackage);
		// }
		// }
		// 主动模式监测到不是所有的channel发送成功
		if (!sendMsgIsNormal)
		{
			// dispatcher.addMessage(dataPackage);
			// dispatcher.setSendable(false);
			logger.warn("单元接口[" + dispatcher.getRealStreamUnit().getOnlyUnitCode() + "]监测到数据没有可以发送的通道 停止发送");
		}
	}

}
